package cn.doitedu.dao;

import cn.doitedu.pojo.CombineCondition;
import cn.doitedu.pojo.EventBean;
import cn.doitedu.pojo.EventParam;
import cn.doitedu.pojo.LogBean;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.java.utils.ParameterTool;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.stream.Collectors;

public class EventStateDaoImpl {

    private ListState<EventBean> listState;


    public void init(ParameterTool parameterTool, ListState<EventBean> listState) throws Exception {
        this.listState = listState;
    }


    public void close() throws Exception {

    }

    public String queryEventStr(LogBean bean, CombineCondition combineCondition) throws Exception {
        //根据条件进行查询
        Iterator<EventBean> iterator = listState.get().iterator();
        List<String> events = new ArrayList<>();
        if(iterator.hasNext()) {
            EventBean eventBean = iterator.next();
            Long startTime = combineCondition.getStartTime();
            Long endTime = combineCondition.getEndTime();
            //根据外部的条件查询，符合时间范围的行为数据
            if (bean.getTimeStamp() >= startTime && eventBean.getTimeStamp() <= endTime) {
                events.add(eventBean.getEventId());
            }
        }
        //将行为字符串，映射成数字
        StringBuffer sb = new StringBuffer();
        //我实际关系的即要匹配的事件
        List<String> targetEvents = combineCondition.getEvents().stream().map(EventParam::getEventId).collect(Collectors.toList());
        //将实际查出来的事件与要匹配的事件完成映射
        for (String event : events) {
            int index = targetEvents.indexOf(event);
            sb.append(index+1);
        }
        return sb.toString();
    }
}
